/*
 * 文件名：AbstractPollableSource.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：napo
 * 修改时间：2014-12-15
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.protocal.realmq;

import java.util.Properties;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.realstream.server.protocal.realmq.lifrcycle.LifecycleAware;
import com.ailk.bigdata.etl.realstream.server.protocal.realmq.lifrcycle.LifecycleState;
import com.google.common.base.Throwables;

/**
 * @description TODO
 * @author [xuwei3]
 * @version [版本号,2014-12-15]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public abstract class AbstractPollableSource implements LifecycleAware,PollableSource
{
	private static Logger logger = LoggerFactory.getLogger(AbstractPollableSource.class);
	private Exception exception;
	private String name;
	private LifecycleState lifecycleState;

	public AbstractPollableSource()
	{
		lifecycleState = LifecycleState.IDLE;
	}

	public synchronized void configure(Properties properties)
	{
		if (isStarted())
		{
			throw new IllegalStateException("Configure called when started");
		}
		else
		{
			try
			{
				exception = null;
				setLifecycleState(LifecycleState.IDLE);
				doConfigure(properties);
			}
			catch (Exception e)
			{
				exception = e;
				setLifecycleState(LifecycleState.ERROR);
				Throwables.propagate(e);
			}
		}
	}

	public synchronized void start()
	{
		if (exception != null)
		{
			logger.error(String.format("Cannot start due to error: name = %s", getName()), exception);
		}
		else
		{
			try
			{
				doStart();
				setLifecycleState(LifecycleState.START);
			}
			catch (Exception e)
			{
				logger.error(String.format("Unexpected error performing start: name = %s", getName()), e);
				exception = e;
				setLifecycleState(LifecycleState.ERROR);
			}
		}
	}

	public synchronized void stop()
	{
		try
		{
			doStop();
			setLifecycleState(LifecycleState.STOP);
		}
		catch (Exception e)
		{
			logger.error(String.format("Unexpected error performing stop: name = %s", getName()), e);
			setLifecycleState(LifecycleState.ERROR);
		}
	}

	public synchronized void setName(String name)
	{
		this.name = name;
	}

	public synchronized String getName()
	{
		return name;
	}

	public synchronized LifecycleState getLifecycleState()
	{
		return lifecycleState;
	}

	public String toString()
	{
		return this.getClass().getName() + "{name:" + name + ",state:" + lifecycleState + "}";
	}

	protected boolean isStarted()
	{
		return getLifecycleState() == LifecycleState.START;
	}

	/**
	 * @return Exception thrown during configure() or start()
	 */
	protected Exception getStartException()
	{
		return exception;
	}

	protected synchronized void setLifecycleState(LifecycleState lifecycleState)
	{
		this.lifecycleState = lifecycleState;
	}

	public Status process() throws Exception
	{
		Exception exception = getStartException();
		if (exception != null)
		{
			throw new MessgaeDataAcceptException("Source had error configuring or starting", exception);
		}
		if (!isStarted())
		{
			throw new RuntimeException("Source is not started");
		}
		return doProcess();
	}

	protected abstract void doConfigure(Properties properties) throws MessgaeDataAcceptException;

	protected abstract void doStart() throws MessgaeDataAcceptException;

	protected abstract void doStop() throws MessgaeDataAcceptException;

	protected abstract Status doProcess() throws Exception;
}
